Dubbo Feature - 03. 连接模型

Foreword

请先了解

本章内容针对DubboProtocol.

连接数

默认情况下, 一个Consumer和一个指定的Provider之间只会建立一个连接(无论两者之间有多少个服务调用).

在Linux平台中, 可以通过netstat命令查看两者之间的连接. 下方栗子为在同一机器上运行Provider, Consumer的结果:

tcp6       0      0 :::48707                :::*                    LISTEN      32753/java		-> Provider
tcp6       0      0 10.1.5.54:48806         10.1.5.54:48707         ESTABLISHED 363/java		-> Consumer
tcp6       0      0 10.1.5.54:55894         10.1.5.76:48707         ESTABLISHED 363/java
tcp6       0      0 10.1.5.54:48707         10.1.5.54:48806         ESTABLISHED 32753/java

服务端的端口是配置中指定的, 而Consumer端开启的端口是一个随机的临时端口. 如果有同一机器上有多个Dubbo Consumer和Provider实例的情况下, 偶尔会遇到端口被占用的情况, 即为临时端口的缘故.

连接的建立

连接的建立和前文的ProtocolFilterWrapper.buildInvokerChain在同一链路上. 起于DubboProtocol.refer中, 其中调用了getClients(), 获取到了ExchangeClient[], 用来实例化DubboInvoker :

public <T> Invoker<T> refer(Class<T> serviceType, URL url) throws RpcException {
     // create rpc invoker.
     DubboInvoker<T> invoker = new DubboInvoker<T>(serviceType, url, getClients(url), invokers);
     invokers.add(invoker);
     return invoker;
 }

getClients方法注释中有说明:

If not configured, connection is shared, otherwise, one connection for one service

当然了, 即一个Provider的一个服务建立一个连接. DubboProtocol中通过referenceClientMap来持有共享的ReferenceCountExchangeClient实例, key为URL如10.1.1.1:20880. 默认情况下一个URL只会调用getSharedClient获取一个共享的实例, 如果用户有配置connections, 则会调用多次initClient, 建立多个连接.

更多: ReferenceCountExchangeClient, ghostClientMap

延迟连接

继续观察initClient方法:

if (url.getParameter(Constants.LAZY_CONNECT_KEY, false)) {
	client = new LazyConnectExchangeClient(url, requestHandler);
} else {
	client = Exchangers.connect(url, requestHandler);
}

可以看到如果配置了lazy = true, 则实际上并没有直接建立连接, 而是返回LazyConnectExchangeClient. 它也是ExchangeClient 的实现类. 实现延迟的方式实际上只是覆盖request,send方法, 在其中调用initClient, 判断client为空才建立连接. 建立连接也是调用上文代码else分支的这一句: client = Exchangers.connect(url, requestHandler);

从`Exchangers`到具体连接建立

Exchangers.connect(URL, ExchangeHandler)
	-> HeaderExchanger.connect(URL, ExchangeHandler)
		-> Transpoters.connect(URL, ChannelHandler)
			-> NettyTransporter.connect(ChannelHandler)
				-> NettyClient.connect()

服务端连接控制

上面说明的都是IO的客户端内容, 服务端的连接控制acceptes是在客户端与其连接建立之后发生的:

1 2	NettyHandler.channelConnnected -> NettyServer(AbstraceServer).connected

其中直接判断, 如果连接数量大于指定数量, 即将连接关闭:

if (accepts > 0 && channels.size() > accepts) {
    logger.error("Close channel " + ch + ", cause: The server " + ch.getLocalAddress() + " connections greater than max config " + accepts);
    ch.close();
    return;
}

心跳

Dubbo中有实现应用层的心跳机制. 在Consumer(HeaderExchangeClient)端和Provider(HeaderExchangeServer)端都会开启一个HeartBeatTask.

Consumer端如下:

DubboProtocol.refer
	DubboProtocol.initClient
		Exchangers.connect
			HeaderExchanger.connect
				HeaderExchangeClient.startHeartbeatTimer

而在Provider端export和Consumer端refer之后也都会实例化一个HeartbeatHandler.

1
2
3

DubboProtocol.export
	-> ChannelHandlers.wrap
		-> HeartbeatHandler<init>

整体流程

观察HeartbeatHandler, 除了received方法之外的其他方法, 基本只是额外加了对于最后读写时间的记录.
两端都会开启HeartbeatTask, 默认每隔一分钟(heartbeat)执行一次. 运行时, 对所有活跃的渠道遍历
- 如果最后读取写时间距今超过一分钟(heartbeat), 则向Channel中发送一个event为Request.HEARTBEAT_EVENT的Request.
- 如果最后读取时间距今超过三分钟((heartbeat timeout), 则如果当前实例为Consumer, 则尝试重新连接, 如果为Provider端, 则关闭连接.
继续看HeartbeatHandler.recieved
- 如果收到心跳请求, 则回复一个心跳响应.
- 如果收到心跳响应, 则直接忽略.

值得注意的是, 由于Consumer和Provider端都开启了心跳调度线程, 所以在空闲时哪端先发起心跳是不确定的.

心跳的作用

文档中有对heartbeat属性做简要说明:

心跳间隔，对于长连接，当物理层断开时，比如拔网线，TCP的FIN消息来不及发送，对方收不到断开事件，此时需要心跳来帮助检查连接是否已断开

其中已经说明了FIN消息的不可靠性, 不仅拔网线, 系统层次的崩溃也会导致FIN消息无法发送. 如果有尝试写过Netty的channelInactive方法Demo, 可能会发现, 突然断网是无法触发的.

那么协议层的keepAlive呢? 翻过的资料相比应用层的心跳缺点如下:

心跳的作用不止是了解两端是否连接活跃, 更重要的是判断对方功能是否正常.
TCP协议的keepalive不能穿过代理服务器(TCP keepalive would only check the connection up to the proxy and not the end-to-end connection).

综合上面两点看来, 心跳的用处是判断对方是否能够提供正常功能. 而实际落实到Dubbo的语境下, 似乎不止于此.

// TODO 没有ZK的情况下, 实际上拔线或者kill程序, API端并没有立即反应, 等到调用时才报错.

Netty Configuration

针对老版本Netty, 并非Netty4, 但实现类似.

NettyServer和NettyClient的bossExecutor和workerExecutor都是Executors.newCachedThreadPool.

然而对于Netty线程模型并不熟悉, 不能对这一点发表更多意见. - -||

NettyClient

此外NettyClient还有一些额外配置 :

-  `keepAlive`: 上面已经提到过
- `connectTimeoutMillis`
- `tcpNoDelay`: 防止消息没有及时发送. 详见下方参考资料.